
/*
 * Copyright © 2021 https://www.cestc.cn/ All rights reserved.
 */

package com.zx.learn.flink.restart;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * 状态恢复和重启策略
 * 状态恢复： 从checkpoint 备份的HDFS 将数据 state 恢复到当前程序中
 * 重启策略： 当前程序出现异常，不同时间、重启次数等方式进行程序重启
 *
 * 重启策略的分类：
 * 没有重启 noStrategy
 * 一直重启 （默认）
 * 固定延迟重启策略 ， 重启3次，每次之间延时时间5s
 * 失败率重启策略， 5分钟之内， 重启5次， 每次1分钟
 *
 * Savepoint 手动重启并恢复
 * 手动备份和恢复 *
 * flink savepoint jobid 存储路径
 *
 * 从 savepoint 恢复数据
 * flink run -s 存储路径 --class 全路径类名 --jar jar包
 *
 * 在flink-conf.yml中可以进行配置,示例如下:
 * restart-strategy: fixed-delay
 * restart-strategy.fixed-delay.attempts: 3
 * restart-strategy.fixed-delay.delay: 10 s
 *
 *
 * 从 socket 读取数据，使用 checkpoint 和重启机制， wordcount 统计， 模拟 输入的bug。
 * Date 2021/6/21 11:18
 */
public class CheckpointRestartDemo {
    public static void main(String[] args) throws Exception {
        //1.env 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //===========Checkpoint参数设置====
        //设置Checkpoint的时间间隔为1000ms做一次Checkpoint
        env.enableCheckpointing(1000);
        //设置State状态存储介质 file:///d:/chk
        env.setStateBackend(new FsStateBackend("file:///d:/chk"));
        //new RocksDBStateBackend(checkpointPath)
        //设置两次checkpoint之间的最短间隔时间
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        //设置如果在做Checkpoint过程中出现错误，是否让整体任务失败：true是  false不是
        //当作业被取消时候，保留checkpoint不删除
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig
                .ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //设置checkpointMode为 Exactly_Once 语义
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //设置checkpoint的超时时间,如果 Checkpoint在 60s内尚未完成说明该次Checkpoint失败,则丢弃。
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        //设置同一时间有多少个checkpoint可以同时执行
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

        //=============重启策略===========
        //-1.默认策略:配置了Checkpoint而没有配置重启策略默认使用无限重启
        //-2.配置无重启策略
//        env.setRestartStrategy(RestartStrategies.noRestart());
        //-3.固定延迟重启策略
        //重启3次,每次间隔10s
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 3000));
        //-4.失败率重启--偶尔使用
        //5分钟内重启3次(第3次不包括,也就是最多重启2次),每次间隔10s
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.minutes(5), Time.seconds(10)));
        //2.Source 创建socket数据源
        DataStreamSource<String> source = env.socketTextStream("192.168.88.161", 9999);
        //3.Transformation
        //3.1空格切割出每个单词并flatMap转换成 Tuple2
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = source.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = value.split(" ");
                for (String word : words) {
                    //3.2遍历循环如果遇到 bug 单词就抛出异常模拟程序报错
                    if (word.equals("bug")) {
                        System.out.println("当前程序报错...");
                        throw new Exception("程序 BUG ...");
                    }
                    out.collect(Tuple2.of(word, 1));
                }
            }
        })
                //3.3分组
                //注意:批处理的分组是groupBy,流处理的分组是keyBy
                .keyBy(t -> t.f0)
                //3.4聚合
                .sum(1);

        //4.sink 打印输出
        result.print();
        //5.execute
        env.execute();
    }
}
